package org.xyf.IM.config;

import com.fasterxml.jackson.databind.ObjectMapper;
import jakarta.annotation.PreDestroy;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;

import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.listener.KafkaMessageListenerContainer;
import org.springframework.kafka.listener.MessageListener;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;
import org.xyf.IM.common.CacheData;
import org.xyf.IM.handler.WebsocketHandler;
import org.xyf.IM.pojo.ChatMessage;

import java.util.Map;
import java.util.Objects;


@Slf4j
@Component
public class KafkaConsumerConfig implements ApplicationRunner {


    @Resource
    private KafkaProperties kafkaProperties;

    @Resource
    private ObjectMapper objectMapper;

    private KafkaMessageListenerContainer<String, Object> container;


    //后续改为ConcurrentMessageListenerContainer
    @Override
    public void run(ApplicationArguments args) throws Exception {
        String topicName = CacheData.getTopicName();
        ContainerProperties containerProps = new ContainerProperties(topicName);//指定要监听的topic
        containerProps.setMessageListener(new MessageListener<String, String>() {
            @Override
            public void onMessage(ConsumerRecord<String, String> data) {
                dealMsg(data);
            }
        });
        Map<String, Object> consumerProps = kafkaProperties.buildConsumerProperties();
//        consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG,)
        DefaultKafkaConsumerFactory<String, Object> cf =
                new DefaultKafkaConsumerFactory<>(consumerProps);
        KafkaMessageListenerContainer<String, Object> container =
                new KafkaMessageListenerContainer<>(cf, containerProps);
        this.container = container;
        log.info("开始监听topic:{}",topicName);
        container.start();
    }

    public void subscribe(String topicName, Map<String, Object> properties) {


        //在新线程中执行订阅
//        executorService.submit(() -> {
//                    log.info("正在开始订阅topic:{}", topicName);
//                    Consumer<String, Object> consumer = null;
//                    try {
//                        consumer = new KafkaConsumer<>(properties);
//                    } catch (Exception e) {
//                        log.error("订阅topic时发生了错误 error is :",e);
//                        SpringApplication.exit(applicationContext);//优雅停机
//                    }
//                    consumer.subscribe(Collections.singleton(topicName));
//                    while (true) {
//                        ConsumerRecords<String, Object> records;
//                        try {
//                            records = consumer.poll(Duration.ofMillis(TIMEOUT_DURATION));//拉取消息;超时时间100毫秒
//                        } catch (Exception e) { //可能需要发送邮件，马上排查问题
//                            log.error("topic subscribe! error: ", e);
//                            continue;
//                        }
//                        for (ConsumerRecord<String, Object> record : records) {
//                            this.dealMsg(record);
//                        }
//                    }
//                }
//        );

    }

    private void dealMsg(ConsumerRecord<String, String> record) {
        try {
            log.info("订阅到消息: {}", record);
            String value = record.value();
            ChatMessage msgBody = objectMapper.readValue(value,ChatMessage.class);
            WebSocketSession recipientSession = WebsocketHandler.sessionMap.get(msgBody.getRecipient());
            if(Objects.nonNull(recipientSession)){//在线则直接发送消息
                recipientSession.sendMessage(new TextMessage(value));
            } else {
                //TODO 不在线此消息则需要做额外的存储
            }
        } catch (Exception e) {
            log.error("消息处理出错了! record is : {}; and error is : ", record, e);
        }
    }

    @PreDestroy
    public void stopContainer() {
        if(Objects.nonNull(container)){
            log.info("开始停止kafka listener容器!");
            container.destroy();
        }
    }

}
